package drds.binlog.example;

import com.google.protobuf.InvalidProtocolBufferException;
import drds.binlog.client.Connector;
import drds.binlog.protocol.Message;
import drds.binlog.protocol.entry_protocol_proto.*;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.MDC;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * 测试基类
 *
 * @author jianghang 2013-4-15 下午04:17:12
 * @version 1.0.4
 */
public class AbstractCanalClientTest
{

    protected final static Logger logger = LoggerFactory.getLogger(AbstractCanalClientTest.class);
    protected static final String SEP = SystemUtils.LINE_SEPARATOR;
    protected static final String DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    protected static String context_format = null;
    protected static String row_format = null;
    protected static String transaction_format = null;

    static
    {
        context_format = SEP + "****************************************************" + SEP;
        context_format += "* Batch Id: [{}] ,count : [{}] , memsize : [{}] , Time : {}" + SEP;
        context_format += "* Start : [{}] " + SEP;
        context_format += "* End : [{}] " + SEP;
        context_format += "****************************************************" + SEP;

        row_format = SEP
                + "----------------> binlog[{}:{}] , name[{},{}] , eventType : {} , executeTime : {}({}) , gtid : ({}) , delay : {} ms"
                + SEP;

        transaction_format = SEP
                + "================> binlog[{}:{}] , executeTime : {}({}) , gtid : ({}) , delay : {}ms"
                + SEP;

    }

    protected volatile boolean running = false;
    protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler()
    {

        public void uncaughtException(Thread t, Throwable e)
        {
            logger.error("parse events has an error", e);
        }
    };
    protected Thread thread = null;
    protected Connector connector;
    protected String destination;

    public AbstractCanalClientTest(String destination)
    {
        this(destination, null);
    }

    public AbstractCanalClientTest(String destination, Connector connector)
    {
        this.destination = destination;
        this.connector = connector;
    }

    /**
     * 获取当前Entry的 GTID信息示例
     *
     * @param header
     * @return
     */
    public static String getCurrentGtid(Header header)
    {
        List<Pair> props = header.getPropsList();
        if (props != null && props.size() > 0)
        {
            for (Pair pair : props)
            {
                if ("curtGtid".equals(pair.getKey()))
                {
                    return pair.getValue();
                }
            }
        }
        return "";
    }

    /**
     * 获取当前Entry的 GTID Sequence No信息示例
     *
     * @param header
     * @return
     */
    public static String getCurrentGtidSn(Header header)
    {
        List<Pair> props = header.getPropsList();
        if (props != null && props.size() > 0)
        {
            for (Pair pair : props)
            {
                if ("curtGtidSn".equals(pair.getKey()))
                {
                    return pair.getValue();
                }
            }
        }
        return "";
    }

    /**
     * 获取当前Entry的 GTID Last Committed信息示例
     *
     * @param header
     * @return
     */
    public static String getCurrentGtidLct(Header header)
    {
        List<Pair> props = header.getPropsList();
        if (props != null && props.size() > 0)
        {
            for (Pair pair : props)
            {
                if ("curtGtidLct".equals(pair.getKey()))
                {
                    return pair.getValue();
                }
            }
        }
        return "";
    }

    protected void start()
    {
        Assert.notNull(connector, "connector is null");
        thread = new Thread(new Runnable()
        {

            public void run()
            {
                process();
            }
        });

        thread.setUncaughtExceptionHandler(handler);
        running = true;
        thread.start();
    }

    protected void stop()
    {
        if (!running)
        {
            return;
        }
        connector.stopRunning();
        running = false;
        if (thread != null)
        {
            try
            {
                thread.join();
            } catch (InterruptedException e)
            {
                // ignore
            }
        }

        MDC.remove("destination");
    }

    protected void process()
    {
        int batchSize = 5 * 1024;
        while (running)
        {
            try
            {
                MDC.put("destination", destination);
                connector.connect();
                connector.subscribe();
                while (running)
                {
                    Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
                    long batchId = message.getId();
                    int size = message.getEntryList().size();
                    if (batchId == -1 || size == 0)
                    {
                        // try {
                        // Thread.sleep(1000);
                        // } catch (InterruptedException e) {
                        // }
                    } else
                    {
                        printSummary(message, batchId, size);
                        printEntry(message.getEntryList());
                    }

                    connector.ack(batchId); // 提交确认
                    // connector.rollback(batchId); // 处理失败, 回滚数据
                }
            } catch (Exception e)
            {
                logger.error("process error!", e);
            } finally
            {
                connector.disconnect();
                MDC.remove("destination");
            }
        }
    }

    private void printSummary(Message message, long batchId, int size)
    {
        long memsize = 0;
        for (Entry entry : message.getEntryList())
        {
            memsize += entry.getHeader().getEventLength();
        }

        String startPosition = null;
        String endPosition = null;
        if (!CollectionUtils.isEmpty(message.getEntryList()))
        {
            startPosition = buildPositionForDump(message.getEntryList().get(0));
            endPosition = buildPositionForDump(message.getEntryList().get(message.getEntryList().size() - 1));
        }

        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        logger.info(context_format, new Object[]{batchId, size, memsize, format.format(new Date()), startPosition,
                endPosition});
    }

    protected String buildPositionForDump(Entry entry)
    {
        long time = entry.getHeader().getExecuteTime();
        Date date = new Date(time);
        SimpleDateFormat format = new SimpleDateFormat(DATE_FORMAT);
        String position = entry.getHeader().getLogfileName() + ":" + entry.getHeader().getLogfileOffset() + ":"
                + entry.getHeader().getExecuteTime() + "(" + format.format(date) + ")";
        if (StringUtils.isNotEmpty(entry.getHeader().getGtid()))
        {
            position += " gtid(" + entry.getHeader().getGtid() + ")";
        }
        return position;
    }

    protected void printEntry(List<Entry> entrys)
    {
        for (Entry entry : entrys)
        {
            long executeTime = entry.getHeader().getExecuteTime();
            long delayTime = new Date().getTime() - executeTime;
            Date date = new Date(entry.getHeader().getExecuteTime());
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND)
            {
                if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN)
                {
                    TransactionBegin begin = null;
                    try
                    {
                        begin = TransactionBegin.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e)
                    {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务头信息，执行的线程id，事务耗时
                    logger.info(transaction_format,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                    entry.getHeader().getGtid(), String.valueOf(delayTime)});
                    logger.info(" BEGIN ----> Thread id: {}", begin.getThreadId());
                    printXAInfo(begin.getPropsList());
                } else if (entry.getEntryType() == EntryType.TRANSACTIONEND)
                {
                    TransactionEnd end = null;
                    try
                    {
                        end = TransactionEnd.parseFrom(entry.getStoreValue());
                    } catch (InvalidProtocolBufferException e)
                    {
                        throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                    }
                    // 打印事务提交信息，事务id
                    logger.info("----------------\n");
                    logger.info(" END ----> transaction id: {}", end.getTransactionId());
                    printXAInfo(end.getPropsList());
                    logger.info(transaction_format,
                            new Object[]{entry.getHeader().getLogfileName(),
                                    String.valueOf(entry.getHeader().getLogfileOffset()),
                                    String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                    entry.getHeader().getGtid(), String.valueOf(delayTime)});
                }

                continue;
            }

            if (entry.getEntryType() == EntryType.ROWDATA)
            {
                RowChange rowChage = null;
                try
                {
                    rowChage = RowChange.parseFrom(entry.getStoreValue());
                } catch (Exception e)
                {
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                }

                EventType eventType = rowChage.getEventType();

                logger.info(row_format,
                        new Object[]{entry.getHeader().getLogfileName(),
                                String.valueOf(entry.getHeader().getLogfileOffset()), entry.getHeader().getSchemaName(),
                                entry.getHeader().getTableName(), eventType,
                                String.valueOf(entry.getHeader().getExecuteTime()), simpleDateFormat.format(date),
                                entry.getHeader().getGtid(), String.valueOf(delayTime)});

                if (eventType == EventType.QUERY || rowChage.getIsDdl())
                {
                    logger.info(" sql ----> " + rowChage.getSql() + SEP);
                    continue;
                }

                printXAInfo(rowChage.getPropsList());
                for (RowData rowData : rowChage.getRowDatasList())
                {
                    if (eventType == EventType.DELETE)
                    {
                        printColumn(rowData.getBeforeColumnsList());
                    } else if (eventType == EventType.INSERT)
                    {
                        printColumn(rowData.getAfterColumnsList());
                    } else
                    {
                        printColumn(rowData.getAfterColumnsList());
                    }
                }
            }
        }
    }

    protected void printColumn(List<Column> columns)
    {
        for (Column column : columns)
        {
            StringBuilder builder = new StringBuilder();
            try
            {
                if (StringUtils.containsIgnoreCase(column.getMysqlType(), "BLOB")
                        || StringUtils.containsIgnoreCase(column.getMysqlType(), "BINARY"))
                {
                    // get value bytes
                    builder.append(column.getName() + " : "
                            + new String(column.getValue().getBytes("ISO-8859-1"), "UTF-8"));
                } else
                {
                    builder.append(column.getName() + " : " + column.getValue());
                }
            } catch (UnsupportedEncodingException e)
            {
            }
            builder.append("    type=" + column.getMysqlType());
            if (column.getUpdated())
            {
                builder.append("    update=" + column.getUpdated());
            }
            builder.append(SEP);
            logger.info(builder.toString());
        }
    }

    protected void printXAInfo(List<Pair> pairs)
    {
        if (pairs == null)
        {
            return;
        }

        String xaType = null;
        String xaXid = null;
        for (Pair pair : pairs)
        {
            String key = pair.getKey();
            if (StringUtils.endsWithIgnoreCase(key, "XA_TYPE"))
            {
                xaType = pair.getValue();
            } else if (StringUtils.endsWithIgnoreCase(key, "XA_XID"))
            {
                xaXid = pair.getValue();
            }
        }

        if (xaType != null && xaXid != null)
        {
            logger.info(" ------> " + xaType + " " + xaXid);
        }
    }

    public void setConnector(Connector connector)
    {
        this.connector = connector;
    }

}
